package transpond

import (
	"encoding/json"
	bolt "go.etcd.io/bbolt"
	"net/http"
	"time"
)

// -----------------------------------------------------------------------------
// 数据结构
// -----------------------------------------------------------------------------
type PendingRequest struct {
	ID                  string      // 消息ID，全局唯一
	URL                 string      // 目标URL
	Method              string      // HTTP方法
	Headers             http.Header // 原始请求头
	Body                []byte      // 原始请求体
	Retry               int         // 重试次数
	NextTry             time.Time   // 下一次尝试时间
	Status              string      `json:"status"`                // pending: 待处理, processing: 处理中
	ProcessingStartTime time.Time   `json:"processing_start_time"` // 处理开始时间（用于超时判断）
}

// -----------------------------------------------------------------------------
// 本地持久化队列（BoltDB）
// -----------------------------------------------------------------------------
type Queue struct {
	db                *bolt.DB
	processingTimeout time.Duration // 处理超时阈值（超过此时间未完成则自动恢复）

}

func NewQueue(path string) (*Queue, error) {
	db, err := bolt.Open(path, 0600, &bolt.Options{Timeout: 1 * time.Second})
	if err != nil {
		return nil, err
	}

	// 创建存储桶
	if err := db.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists([]byte("requests"))
		return err
	}); err != nil {
		db.Close()
		return nil, err
	}

	q := &Queue{
		db:                db,
		processingTimeout: 5 * time.Minute, // 默认处理超时5分钟，可根据业务调整
	}

	// 崩溃恢复：将超时的"处理中"请求恢复为"待处理"
	if err := q.recoverProcessingRequests(); err != nil {
		q.Close()
		return nil, err
	}

	return q, nil
}

// （新增）崩溃恢复：扫描超时的"处理中"请求，恢复为"待处理"
func (q *Queue) recoverProcessingRequests() error {
	return q.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte("requests"))
		c := b.Cursor()
		for k, v := c.First(); k != nil; k, v = c.Next() {
			var req PendingRequest
			if err := json.Unmarshal(v, &req); err != nil {
				_ = b.Delete(k) // 损坏数据直接删除
				continue
			}

			// 若状态为"处理中"，直接恢复为"待处理"（删除超时判断）
			if req.Status == "processing" {
				req.Status = "pending"
				req.ProcessingStartTime = time.Time{} // 清空处理时间
				val, _ := json.Marshal(req)
				if err := b.Put(k, val); err != nil {
					return err
				}
			}
		}
		return nil
	})
}

func (q *Queue) Close() error { return q.db.Close() }

// 入队（事务写入）
func (q *Queue) Enqueue(req *PendingRequest) error {
	return q.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte("requests"))
		key := []byte(req.ID)
		val, _ := json.Marshal(req)
		return b.Put(key, val)
	})
}

// Dequeue 取出待处理请求（修改为状态标记，不直接删除）
func (q *Queue) Dequeue(n int) []*PendingRequest {
	var out []*PendingRequest
	_ = q.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte("requests"))
		c := b.Cursor()
		now := time.Now()

		for k, v := c.First(); k != nil && len(out) < n; k, v = c.Next() {
			var req PendingRequest
			if err := json.Unmarshal(v, &req); err != nil {
				_ = b.Delete(k) // 损坏数据直接删除
				continue
			}

			// 只处理"待处理"且到达下次尝试时间的请求 有可能status 是空字符串也要处理
			if req.Status == "" {
				req.Status = "pending"
			}
			if req.Status != "pending" || req.NextTry.After(now) {
				continue
			}

			// 标记为"处理中"，记录开始时间（原子更新）
			req.Status = "processing"
			req.ProcessingStartTime = now
			val, _ := json.Marshal(req)
			if err := b.Put(k, val); err != nil {
				continue // 更新失败则跳过，下次重试
			}

			out = append(out, &req)
		}
		return nil
	})
	return out
}

// Update 更新请求状态（重试时使用，恢复为"待处理"）
func (q *Queue) Update(req *PendingRequest) error {
	// 重试时将状态改回"待处理"，清空处理时间
	req.Status = "pending"
	req.ProcessingStartTime = time.Time{}
	return q.Enqueue(req)
}

// 删除（成功确认）
func (q *Queue) Delete(id string) error {
	return q.db.Update(func(tx *bolt.Tx) error {
		return tx.Bucket([]byte("requests")).Delete([]byte(id))
	})
}
